package cn.com.infcn.util.mss;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ExecutionException;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.time.FastDateFormat;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.nlpcn.commons.lang.util.ObjConver;
import org.nlpcn.commons.lang.util.StringUtil;
import org.nutz.mvc.annotation.Param;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Repository;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.ImmutableMap;

import cn.com.infcn.entities.MssField;
import cn.com.infcn.entities.MssTable;
import cn.com.infcn.po.Restful;
import cn.com.infcn.repositories.MssFieldDao;
import cn.com.infcn.repositories.MssTableDao;

/**
 * 创建索引api
 *
 * @author Ansj
 */
@Repository
public class MssDataApi {


	@Autowired
	private TransportClient mssEsClient;
	@Autowired
	private MssTableDao mssTableDao;
	@Autowired
	private MssFieldDao mssFieldDao;
	
	@Value("${mss_db}")
	private String mss_db;


	private static final String KEY_NAME = "_id";
	private static final String ERROR_FILED_NAME = "_err";
	private static final String FIELD_INSERT_TIME = "_insert_time";

	/**
	 * 创建索引,
	 *
	 * @param name 库表名称
	 * @param jsonStr 对象json格式
	 * @param skiperr 是否跳过错误
	 * @param operation 操作类型(默认是replace): replace(覆盖), update(更新)
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	public Object insert(@Param("name") final String name, @Param("jsonStr") String jsonStr, @Param(value = "skiperr", df = "true") boolean skiperr,
			@Param(value = "operation", df = "replace") String operation) throws InterruptedException, ExecutionException {

		MssTable tableInfo = mssTableDao.findByName(mss_db);
		if (tableInfo!=null) {
			List<MssField> findByTableId = mssFieldDao.findByTableId(tableInfo.getId());
			tableInfo.setFields(findByTableId);
		}
		
		Object parse = JSON.parse(jsonStr);

		JSONArray arr;
		if (parse instanceof JSONObject) {
			arr = new JSONArray();
			arr.add(parse);
		} else {
			arr = (JSONArray) parse;
		}

		BulkRequestBuilder prepareBulk = mssEsClient.prepareBulk() ;

		Map<String,Object> doc;
		String id;
		List<Object> errors = new ArrayList<>(arr.size());
		String errMessage;
		boolean isUpdate = "update".equalsIgnoreCase(operation);

		for (Object object : arr) {
			doc = null;
			try {
				doc = md(tableInfo, (JSONObject) object);
			} catch (Exception e) {
				errMessage = ExceptionUtils.getStackTrace(e);

//				LOG.error("invalid data[" + object + "]: " + errMessage);
				if (skiperr) {
					((JSONObject) object).put(ERROR_FILED_NAME, e.getMessage());
					errors.add(object);
				} else {
					return Restful.instance(false, "data format err , all record to skip ", ImmutableMap.of("err_detail", errMessage), 500);
				}
			}

			if (doc != null) {
				id = doc.remove(KEY_NAME).toString();

				 if (isUpdate) {
                     prepareBulk.add(mssEsClient.prepareUpdate(name, name, id).setDoc(doc).setDocAsUpsert(true));
                 } else {
                     prepareBulk.add(mssEsClient.prepareIndex(name, name, id).setSource(doc));
                 }
			}
		}


		// 索引更新
		try {
			prepareBulk.get();
		} catch (Exception e) {
			errMessage = ExceptionUtils.getStackTrace(e);
//			LOG.error("bulk[" + arr.size() + "] to es error: " + errMessage);
			return Restful.instance(false, "save to index err ", ImmutableMap.of("err_detail", errMessage, "errors", errors), 502);
		}

		return Restful.instance(ImmutableMap.of("errors", errors));
	}





	/**
	 * 构建文档对象进行校验
	 *
	 * @param job
	 * @return
	 * @
	 */
	private Map<String,Object> md(MssTable tableInfo, Map<String, Object> job) {

		// 设置库里的主键值为字符串类型
		String key = Optional.ofNullable(job.get(KEY_NAME)).orElse("").toString();
		if (StringUtil.isBlank(key)) {
			key = UUID.randomUUID().toString();
		}

		if (tableInfo.getStrict()) {
			// 如果锁定表结构, 就只取定义的字段
			JSONObject json = new JSONObject();
			for (MssField field : tableInfo.getFields()) {
				if (job.containsKey(field.getName())) {
					json.put(field.getName(), getObjectByField(job.get(field.getName()), field));
				}
			}
			job = json;
		} else {
			// 如果不锁定, 映射定义的字段并且保留其他字段
			for (MssField field : tableInfo.getFields()) {
				if (job.containsKey(field.getName())) {
					job.put(field.getName(), getObjectByField(job.get(field.getName()), field));
				}
			}
		}

		job.putIfAbsent(FIELD_INSERT_TIME, new Date());
		job.put(KEY_NAME, key);

		return job;
	}

	/**
	 * 根据对象格式转换对象
	 *
	 * @param value
	 * @param field
	 * @
	 */
	private Object getObjectByField(Object value, MssField field)  {

		if (value == null) {
			return null;
		}

		// TODO: 如果是数组类型, 直接返回
		if (value instanceof Collection) {
			return value;
		}

		switch (field.getType()) {
		case "string":
			return StringUtils.isEmpty(field.getSplit()) ? String.valueOf(value) : Arrays.asList(String.valueOf(value).split(field.getSplit()));
		case "ip":
		case "completion":
			return value;
		case "date":
			return value instanceof Date ? value : ObjConver.castToDate(value);
		case "long":
			return ObjConver.castToLong(value);
		case "double":
			return ObjConver.castToDouble(value);
		case "boolean":
			return ObjConver.castToBoolean(value);
		case "object":
			return value;
		case "nested":
		case "geo_point":
		case "geo_shape":
			return JSONObject.toJSON(value);
		default:
			return JSONObject.toJSON(value);//TODO
//			throw new Exception(505, "unknown type " + field.getType());
		}
	}


	/**
	 * 根据id删除某条数据
	 *
	 * @param name 库表名称
	 * @param id 数据id
	 * @return
	 * @
	 */
	public Object delete(String name, String id)  {
		// 索引删除
		DeleteResponse result = mssEsClient.prepareDelete(name, name, id).get();
//		LOG.info(String.format("delete %s in es[%s]: {found: %s}", id, name, result.isFound()));
		return Restful.instance(result);
	}



//	public static void main(String[] args) throws Exception {
//		Testing.instance(MssDataApi.class).reindex("bbbb", null, false);
//	}
}
